package com.yanggu.flink.datastream_api.offlinealarm

import cn.hutool.core.date.DateUtil
import cn.hutool.core.util.NumberUtil
import cn.hutool.json.JSONUtil
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor, ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessAllWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import java.util.Date
import scala.beans.BeanProperty
import scala.collection.mutable.ListBuffer

object AlarmDemo03 {

  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)

    val windowDataStream = environment
      .readTextFile(getClass.getResource("/CEP_Test").getPath)
      .flatMap((line, out: Collector[CEPData]) => out.collect(JSONUtil.toBean(line, classOf[CEPData])))
      .assignAscendingTimestamps(_.etime)
      .windowAll(TumblingEventTimeWindows.of(Time.seconds(1L)))
      .process(new ProcessAllWindowFunction[CEPData, WindowData, TimeWindow] {
        override def process(context: Context, elements: Iterable[CEPData], out: Collector[WindowData]): Unit = {
          val start = context.window.getStart
          val end = context.window.getEnd
          val size2 = elements.count(_.status != 200)
          val size = elements.size
          out.collect(WindowData(start, end, NumberUtil.div(size2, size)))
          println(s"在${DateUtil.formatDateTime(new Date(start))}到${DateUtil.formatDateTime(new Date(end))}" +
            s", 非200的比例${NumberUtil.div(size2, size)}")
        }
      })
    val dataStream = windowDataStream
      .keyBy(data => true)
      .process(new KeyedProcessFunction[Boolean, WindowData, String] {

        private var alarmFlag: ValueState[Boolean] = _

        private var alarmList: ListState[WindowData] = _

        override def open(parameters: Configuration): Unit = {
          alarmFlag = getRuntimeContext.getState(new ValueStateDescriptor[Boolean]("alarmFlag", classOf[Boolean], false))
          alarmList = getRuntimeContext.getListState(new ListStateDescriptor[WindowData]("alarmList", classOf[WindowData]))
        }

        override def processElement(value: WindowData, ctx: KeyedProcessFunction[Boolean, WindowData, String]#Context, out: Collector[String]) = {
          if (value.ratio > 0.4) {
            alarmList.add(value)

            val datas = new ListBuffer[WindowData]()
            alarmList.get().forEach(data => datas += data)

            if (datas.size >= 3) {
              alarmFlag.update(true)
              val buffer = new StringBuffer()
              datas.takeRight(3).foreach(data => {
                val str = s"在${DateUtil.formatDateTime(new Date(data.windowStart))}到${DateUtil.formatDateTime(new Date(data.windowEnd))}, 非200的比例${data.ratio}, "
                buffer.append(str)
              })
              ctx.output(new OutputTag[String]("alarm"), s"${buffer.toString.dropRight(1)}")
            }
          } else {
            alarmList.clear()
            if (alarmFlag.value()) {
              alarmFlag.update(false)
              ctx.output(new OutputTag[String]("resume"), s"在${DateUtil.formatDateTime(new Date(value.windowEnd))}, 恢复告警")
            }
          }
        }

      })

    dataStream.getSideOutput(new OutputTag[String]("resume")).print("resume")
    dataStream.getSideOutput(new OutputTag[String]("alarm")).print("alarm")

    environment.execute()

  }

}

case class CEPData(@BeanProperty var id: Int,
                   @BeanProperty var status: Int,
                   @BeanProperty var etime: Long)

case class WindowData(windowStart: Long,
                      windowEnd: Long,
                      ratio: Double)